package server;

import com.google.protobuf.ByteString;
import com.google.protobuf.InvalidProtocolBufferException;
import common.BlockInfo;
import common.FileInfo;
import common.StateCode;
import common.Utils;
import hdfs.hdfsPut;
import interfaces.ClientDef;
import interfaces.DataNodeDef;
import interfaces.NameNodeDef;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import transport.ClientTransPorter;

import java.io.*;
import java.rmi.Naming;
import java.rmi.RemoteException;
import java.util.ArrayList;
import java.util.List;

import static common.StateCode.*;
import static common.Utils.*;


public class ClientImpl implements ClientDef {
    private static final Logger logger = LoggerFactory.getLogger(ClientImpl.class);
    private String NN_IP;
    private int NN_PORT;
    private PrintWriter logWriter;
    private NameNodeDef nameNodeStub;
    private boolean isLocal;

    public ClientImpl(String NN_IP, int NN_PORT) {
        this.NN_IP = NN_IP;
        this.NN_PORT = NN_PORT;
        this.isLocal = false;
        init();
        logger.info("NameNode初始化完成");
    }

    public ClientImpl(String NN_IP, int NN_PORT, boolean isLocal) {
        this.NN_IP = NN_IP;
        this.NN_PORT = NN_PORT;
        this.isLocal = isLocal;
        init();
    }


    private void init() {
        logger.info("NameNode初始化...");
        try {
            getNNStub(NN_IP, NN_PORT);
        } catch (Exception e) {
            logger.error("NameNode初始化失败");
            e.printStackTrace();
        }
    }

    private void getNNStub(String NN_IP, int NN_PORT) {
        try {
            String location = "//" + NN_IP + ":" + NN_PORT + "/NN";
            System.out.println("trying get NNStub" + location);
            nameNodeStub = (NameNodeDef) Naming.lookup(location);
        } catch (Exception e) {
            e.printStackTrace();
        }
    }


    /**
     * 1.openFile RPC 向nameNode 确认是否可以上传 nameNode会在元信息中增加一个空文件
     * 2.对文件分块：依次读取block
     * 3.对每一个block： assignBlock RPC 从nameNode获取存储位置 nameNode 会在元信息中增加一个空块
     *
     * @param remotePath 存储路径
     * @param localPath  文件名
     * @return
     * @throws RemoteException
     */
    private List<BlockInfo> put(String remotePath, String localPath) throws RemoteException {
        hdfsPut.OpenFileRequest.Builder openFileRequest = hdfsPut.OpenFileRequest.newBuilder();
        openFileRequest.setFileName(remotePath);
        openFileRequest.setForRead(false);
        ArrayList<BlockInfo> ans = new ArrayList<>();
        try {
            // 先询问是否可以上传
            byte[] inp = nameNodeStub.openFile(openFileRequest.build().toByteArray());
            hdfsPut.OpenFileResponse response = hdfsPut.OpenFileResponse.parseFrom(inp);
            if (response.getStatus() == StateCode.fileExist) {
                logger.error("已存在同名文件");
                return new ArrayList<>();
            }
            logger.info("上传文件" + remotePath);
            // 获得文件切片
            File file = new File(localPath);
            //buffer大小固定为BLOCK_SIZE,之后再加上4位的crc校验位
            byte[] buffer = new byte[BLOCK_SIZE];
            FileInputStream finStream = new FileInputStream(file);
            int len = 0;
            BufferedInputStream bufStream = new BufferedInputStream(finStream);
            hdfsPut.AssignBlockRequest.Builder assignReq = hdfsPut.AssignBlockRequest.newBuilder();
            assignReq.setFilename(remotePath);
            while ((len = bufStream.read(buffer, 0, BLOCK_SIZE)) > 0) {
                // 获取文件块位置
                logger.info("文件分块大小为：" + len);
                assignReq.setSize(len);
                //当发送数据小于BLOCK_SIZE时缩减buffer大小
                if (len < BLOCK_SIZE) {
                    byte[] temp = new byte[len];
                    System.arraycopy(buffer, 0, temp, 0, len);
                    buffer = temp;
                }
                byte[] out = nameNodeStub.assignBlock(assignReq.build().toByteArray());
                hdfsPut.AssignBlockResponse assignRes = hdfsPut.AssignBlockResponse.parseFrom(out);
                hdfsPut.BlockLocations blockLocations = assignRes.getNewBlock();
//                logger.info("块位置");
//                for (hdfsPut.DataNodeLocation dataNodeLocation : blockLocations.getLocationsList()) {
//                    logger.info(dataNodeLocation.getPort()+" ");
//                }
//                logger.info("向dataNode中写块");
//                out = writeBlock(blockLocations, buffer,len);
//                hdfsPut.WriteBlockResponse writeRes = hdfsPut.WriteBlockResponse.parseFrom(out);
//                // TODO: 更合理的失败重传机制
//                while(writeRes.getStatus()!=blockOk){
//                    out = writeBlock(blockLocations, buffer,len);
//                    writeRes = hdfsPut.WriteBlockResponse.parseFrom(out);
//                }
                while (!writeBlockWithTansPorter(blockLocations, buffer, len))
                    logger.info("block" + blockLocations.getBlockNumber() + "上传成功");
                BlockInfo blockInfo = new BlockInfo(remotePath, blockLocations.getBlockNumber(), len);
                for (hdfsPut.DataNodeLocation dn : blockLocations.getLocationsList()) {
                    blockInfo.appendLocation(dn.getIp(), dn.getPort());
                }
                ans.add(blockInfo);
                //如果len的长度小于BLOCK_SIZE，说明文件读取可以结束了
                if (len < BLOCK_SIZE) {
                    break;
                }
            }

            // 关闭文件
            hdfsPut.CloseFileRequest.Builder closeReq = hdfsPut.CloseFileRequest.newBuilder();
            closeReq.setFileName(remotePath);
            byte[] out = nameNodeStub.closeFile(closeReq.build().toByteArray());
            hdfsPut.CloseFileResponse closeRes = hdfsPut.CloseFileResponse.parseFrom(out);
            if (closeRes.getStatus() == fileOk) {
                logger.info("文件上传成功");
            }
            //文件上传完毕后将数据写入磁盘
            hdfsPut.SynFileRequest.Builder synReq = hdfsPut.SynFileRequest.newBuilder();
            synReq.setFilename(remotePath);
            byte[] bytes = nameNodeStub.SynFile(synReq.build().toByteArray());
            hdfsPut.SynFileResponse synFileResponse = hdfsPut.SynFileResponse.parseFrom(bytes);
            if (synFileResponse.getStatus() != fileOk) {
                logger.error(remotePath + "同步到磁盘失败");
            }
        } catch (IOException e) {
            //TODO：更合理的异常处理
            e.printStackTrace();
        }
        return ans;
    }

    /**
     * 写块
     * 暂时实现为基于protobuf的文件传输
     *
     * @return
     */
    private boolean writeBlock(hdfsPut.BlockLocations blockLocations, byte[] block, int size) throws RemoteException {
        hdfsPut.WriteBlockRequest.Builder request = hdfsPut.WriteBlockRequest.newBuilder();
        request.setReplicate(REPLICATE_NUM);
        request.setBlockInfo(blockLocations);
        request.setSize(size);
        block = unitBytes(block, addCRC(block));
        request.setData(ByteString.copyFrom(block));
        hdfsPut.DataNodeLocation dnLocation = blockLocations.getLocations(0);
        DataNodeDef dnStub = getDNStub(dnLocation.getIp(), dnLocation.getPort());
        assert dnStub != null;
        byte[] out = dnStub.writeBlock(request.build().toByteArray());
        hdfsPut.WriteBlockResponse response = null;
        try {
            response = hdfsPut.WriteBlockResponse.parseFrom(out);
            if (response.getStatus() == blockOk) {
                return true;
            }
        } catch (InvalidProtocolBufferException e) {
            e.printStackTrace();
            return false;
        }
        return false;
    }

    private boolean writeBlockPacket(hdfsPut.BlockLocations blockLocations, byte[] block, int len) throws RemoteException, InvalidProtocolBufferException {
        int offset = 0;
        while (offset < len) {
            hdfsPut.WriteBlockRequest.Builder request = hdfsPut.WriteBlockRequest.newBuilder();
            request.setReplicate(REPLICATE_NUM);
            request.setBlockInfo(blockLocations);
            int size = Math.min(len - offset, PACKET_SIZE);
            request.setData(ByteString.copyFrom(block, offset, size));
            request.setSize(size);
            hdfsPut.DataNodeLocation dnLocation = blockLocations.getLocations(0);
            DataNodeDef dnStub = getDNStub(dnLocation.getIp(), dnLocation.getPort());
            assert dnStub != null;
            byte[] out = dnStub.writeBlock(request.build().toByteArray());
            hdfsPut.WriteBlockResponse response = hdfsPut.WriteBlockResponse.parseFrom(out);
            offset += size;
            if (response.getStatus() != blockOk) {
                return false;
            }
        }
        return true;
    }

    private boolean writeBlockWithTansPorter(hdfsPut.BlockLocations blockLocations, byte[] block, int len) throws RemoteException, InvalidProtocolBufferException {
        hdfsPut.TransPortRequest.Builder request = hdfsPut.TransPortRequest.newBuilder();
        request.setReplicate(REPLICATE_NUM);
        request.setBlockInfo(blockLocations);
        request.setSize(len);
        block = Utils.unitBytes(block, addCRC(block));
        hdfsPut.DataNodeLocation dnLocation = blockLocations.getLocations(0);
        DataNodeDef dnStub = getDNStub(dnLocation.getIp(), dnLocation.getPort());
        assert dnStub != null;
        byte[] out = dnStub.transPort(request.build().toByteArray());
        logger.info("得到transport的结果");
        hdfsPut.TransPortResponse resp = hdfsPut.TransPortResponse.parseFrom(out);
        if (resp.getStatus() == connectOk) {
            ClientTransPorter transPorter = new ClientTransPorter(len + 4);
            logger.info("开始建立连接");
            try {
                if (isLocal) {
                    transPorter.connect("localhost", dnLocation.getTPort());
                } else {
                    transPorter.connect(dnLocation.getIp(), dnLocation.getTPort());
                }
                return transPorter.writeBlock(block, blockLocations.getBlockNumber());
            } catch (InterruptedException e) {
                e.printStackTrace();
                return false;
            }
        }
        return false;
    }

    @Override
    public boolean getFile(String remotePath, String fileName, String localPath) throws RemoteException {
        String SourceFile = remotePath + fileName;
        hdfsPut.OpenFileRequest.Builder request = hdfsPut.OpenFileRequest.newBuilder();
        request.setFileName(SourceFile);
        request.setForRead(true);
        byte[] openFileResponseBytes = nameNodeStub.openFile(request.build().toByteArray());
        byte[] fileBytes = new byte[0];
        boolean simple = true;
        try {
            hdfsPut.OpenFileResponse openFileResponse = hdfsPut.OpenFileResponse.parseFrom(openFileResponseBytes);
            //TODO 返回信息状态
            if (openFileResponse.getStatus() != fileOk) {
                logger.info("没有该文件");
                return false;
            }
            List<Integer> blockNumsList = openFileResponse.getBlockNumsList();
            //遍历block
            for (Integer i : blockNumsList) {
                //根据blockNum获取,向nameNode发送请求获取blockLocations
                hdfsPut.blockLocationRequest.Builder blockRequest = hdfsPut.blockLocationRequest.newBuilder();
                blockRequest.setBlockNum(i);
                byte[] blockLocationsResponseBytes = nameNodeStub.getBlockLocations(blockRequest.build().toByteArray());
                hdfsPut.blockLocationResponse locationResponse = hdfsPut.blockLocationResponse.parseFrom(blockLocationsResponseBytes);
                // 选择一个block，这里总是从第一个dataNode中读取文件
                hdfsPut.DataNodeLocation dataNodeLocation = locationResponse.getDataNodeLocation(0);
                hdfsPut.ReadBlockRequest.Builder readBlockRequest = hdfsPut.ReadBlockRequest.newBuilder();
                readBlockRequest.setBlockNo(i)
                        .setFileName(SourceFile)
                        .setIP(dataNodeLocation.getIp())
                        .setPORT(dataNodeLocation.getPort());
                DataNodeDef dnStub = getDNStub(dataNodeLocation.getIp(), dataNodeLocation.getPort());
                assert dnStub != null;
                //readBlock
                byte[] readBlockResponseBytes = dnStub.readBlock(readBlockRequest.build().toByteArray());
                hdfsPut.ReadBlockResponse readBlockResponse = hdfsPut.ReadBlockResponse.parseFrom(readBlockResponseBytes);
                ByteString data = readBlockResponse.getData();
                if (!simple) {
                    fileBytes = unitBytes(fileBytes, data.toByteArray());
                } else {
                    String temp = write2Temp(data.toByteArray(), i);
                    String toFileName = localPath;
                    File toFile = new File(toFileName);
                    File parentFile = toFile.getParentFile();
                    if (!parentFile.exists()) {
                        parentFile.mkdir();
                    }
                    File fromFile = new File(temp);
                    Utils.fileCopyWithFileChannel(fromFile, toFile);
                    logger.info("读取第" + i + "个block到文件中");
                    if (fromFile.delete()) {
                        logger.info("temp" + i + "已被删除");
                    }
                }
            }
            if (!simple) {
                String destFile = localPath;
                BufferedOutputStream bufferedOutputStream = new BufferedOutputStream(new FileOutputStream(destFile));
                bufferedOutputStream.write(fileBytes);
                bufferedOutputStream.close();
            }
            return true;
        } catch (IOException e) {
            e.printStackTrace();
        }
        return false;
    }

    private String write2Temp(byte[] fileBytes, Integer i) throws IOException {
        String fileName = "temp" + i;
        BufferedOutputStream bis = new BufferedOutputStream(new FileOutputStream("temp" + i));
        bis.write(fileBytes, 0, fileBytes.length);
        bis.close();
        return fileName;
    }

    @Override
    public byte[] getBlock(int blockNo, String remotePath, String fileName, String IP, int PORT) throws RemoteException {
        hdfsPut.ReadBlockRequest.Builder readBlockRequest = hdfsPut.ReadBlockRequest.newBuilder();
        readBlockRequest.setIP(IP)
                .setPORT(PORT)
                .setBlockNo(blockNo)
                .setFileName(remotePath + fileName);
        DataNodeDef dnStub = getDNStub(IP, PORT);
        assert dnStub != null;
        byte[] readBlockBytes = dnStub.readBlock(readBlockRequest.build().toByteArray());
        try {
            hdfsPut.ReadBlockResponse readBlockResponse = hdfsPut.ReadBlockResponse.parseFrom(readBlockBytes);
            if (readBlockResponse.getStatus() != fileOk) {
                return new byte[0];
            }
            return readBlockResponse.getData().toByteArray();
        } catch (InvalidProtocolBufferException e) {
            e.printStackTrace();
        }
        return new byte[0];
    }

    @Override
    public boolean getFile(List<Integer> blockNos, String localPath) throws RemoteException {
        return false;
    }

    @Override
    public List<BlockInfo> put(String remotePath, String fileName, String localPath) throws RemoteException {
        return this.put(remotePath + fileName, localPath);
    }

    @Override
    public List<FileInfo> list() throws RemoteException {
        hdfsPut.AllFileListRequest.Builder builder = hdfsPut.AllFileListRequest.newBuilder();
        byte[] bytes = nameNodeStub.list(builder.build().toByteArray());
        List<FileInfo> fileInfos = new ArrayList<>();
        try {
            hdfsPut.AllFileResponse response = hdfsPut.AllFileResponse.parseFrom(bytes);
            if (response.getStatus() != fileOk) {
                logger.error("读取文件失败");
                return fileInfos;
            }
            List<hdfsPut.FileInfo> infoList = response.getFileInfoList();
            for (hdfsPut.FileInfo fileInfo : infoList) {
                fileInfos.add(new FileInfo(fileInfo.getFileName(),
                        fileInfo.getBlockNumList(), fileInfo.getFileStatus()));
            }
            return fileInfos;
        } catch (InvalidProtocolBufferException e) {
            e.printStackTrace();
        }
        return null;
    }

}
